package com.ruoyi.hadoop.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;

import java.io.IOException;
import java.util.Iterator;

public class HBaseUtils {
    private final static String KEY = "hbase.zookeeper.quorum";
    private final static String VALUE = "bogon:2181";
    private static Configuration configuration;

    static {
        //1. 创建配置对象
        configuration = HBaseConfiguration.create();
        configuration.set(KEY, VALUE);
    }

    public static Admin getAdmin() {
        try {
            Connection connection = ConnectionFactory.createConnection(configuration);
            Admin admin = connection.getAdmin();
            return admin;
        } catch (IOException e) {
            e.printStackTrace();
            return null;
        }
    }

    public static Table getTable() {
        return getTable("user_info");
    }

    public static Table getTable(String tablename) {
        try {
            Connection connection = ConnectionFactory.createConnection(configuration);
            return connection.getTable(TableName.valueOf(tablename));
        } catch (IOException e) {
            e.printStackTrace();
            return null;
        }
    }

    public static void showResult(Result result) throws IOException {
        StringBuffer sb = new StringBuffer();
        CellScanner cellScanner = result.cellScanner();
        while (cellScanner.advance()) {
            Cell cell = cellScanner.current();
            sb.append("Rowkey=").append(new String(CellUtil.cloneRow(cell))).append("\t")
                    .append("columnfamily=").append(new String(CellUtil.cloneFamily(cell))).append("\t")
                    .append("column=").append(new String(CellUtil.cloneQualifier(cell))).append("\t")
                    .append("value=").append(new String(CellUtil.cloneValue(cell))).append("\t").append("\n");
        }
        System.out.println(sb.toString());
    }

    public static void showResult(Filter filter) throws IOException {
        if (filter instanceof SingleColumnValueFilter) {
            SingleColumnValueFilter singleColumnValueFilter = (SingleColumnValueFilter) filter;
            singleColumnValueFilter.setFilterIfMissing(true);
        }

        Scan scan = new Scan();
        scan.setFilter(filter);
        ResultScanner scanner = getTable().getScanner(scan);
        Iterator<Result> iterator = scanner.iterator();
        while (iterator.hasNext()) {
            Result result = iterator.next();
            showResult(result);
        }
    }

    public static void close(Admin admin) {
        try {
            if(admin != null) admin.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void close(Table table) {
        try {
            if(table != null) table.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void close(Admin admin, Table table) {
        close(admin);
        close(table);
    }
}
